package com.datareport.config;

import jakarta.annotation.PostConstruct;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.context.annotation.Configuration;


/**
 * Flink计算配置类
 * 用于加载和管理Flink相关的配置参数
 */
@Slf4j
@Configuration
@ConfigurationProperties(prefix = "flink.compute")
@Data
public class FlinkComputeProperties {

    /**
     * Flink作业配置
     */
    private JobConfig job = new JobConfig();
    
    /**
     * 检查点配置
     */
    private CheckpointConfig checkpoint = new CheckpointConfig();
    
    /**
     * 状态后端配置
     */
    private StateBackendConfig stateBackend = new StateBackendConfig();
    
    /**
     * 重启策略配置
     */
    private RestartConfig restart = new RestartConfig();
    
    /**
     * 窗口配置
     */
    private WindowConfig window = new WindowConfig();
    
    /**
     * Kafka配置
     */
    private KafkaConfig kafka = new KafkaConfig();
    
    /**
     * 数据库配置
     */
    private DatabaseConfig database = new DatabaseConfig();
    
    /**
     * 性能调优配置
     */
    private PerformanceConfig performance = new PerformanceConfig();

    /**
     * 作业配置
     */
    @Data
    public static class JobConfig {
        /**
         * 默认并行度
         */
        private int parallelism = 4;
        
        /**
         * 最大并行度
         */
        private int maxParallelism = 128;
        
        /**
         * 作业超时时间（分钟）
         */
        private int timeoutMinutes = 60;
        
        /**
         * 是否启用自动重启
         */
        private boolean autoRestart = true;
        
        /**
         * 重启间隔（秒）
         */
        private int restartDelaySeconds = 10;
    }

    /**
     * 检查点配置
     */
    @Data
    public static class CheckpointConfig {
        /**
         * 是否启用检查点
         */
        private boolean enabled = true;
        
        /**
         * 检查点间隔（毫秒）
         */
        private long interval = 60000;
        
        /**
         * 检查点超时时间（毫秒）
         */
        private long timeout = 300000;
        
        /**
         * 最小暂停时间（毫秒）
         */
        private long minPauseBetweenCheckpoints = 30000;
        
        /**
         * 最大并发检查点数
         */
        private int maxConcurrentCheckpoints = 1;
        
        /**
         * 是否启用外部化检查点
         */
        private boolean externalizedCheckpointsEnabled = true;
        
        /**
         * 外部化检查点清理模式（DELETE_ON_CANCELLATION/RETAIN_ON_CANCELLATION）
         */
        private String externalizedCheckpointCleanup = "DELETE_ON_CANCELLATION";
    }

    /**
     * 状态后端配置
     */
    @Data
    public static class StateBackendConfig {
        /**
         * 状态后端类型（memory/filesystem/rocksdb）
         */
        private String type = "rocksdb";
        
        /**
         * 状态存储路径
         */
        private String checkpointDir = "file:///tmp/flink-checkpoints";
        
        /**
         * 保存点路径
         */
        private String savepointDir = "file:///tmp/flink-savepoints";
        
        /**
         * 状态大小阈值（字节）
         */
        private long stateSizeThreshold = 1024 * 1024; // 1MB
        
        /**
         * 是否启用增量检查点
         */
        private boolean incrementalCheckpointsEnabled = true;
    }

    /**
     * 重启策略配置
     */
    @Data
    public static class RestartConfig {
        /**
         * 重启策略（fixed-delay/failure-rate/none）
         */
        private String strategy = "fixed-delay";
        
        /**
         * 最大重启次数
         */
        private int maxRestarts = 3;
        
        /**
         * 重启延迟（毫秒）
         */
        private long delayBetweenRestarts = 10000;
        
        /**
         * 故障率重启的时间间隔（毫秒）
         */
        private long failureInterval = 60000;
        
        /**
         * 故障率重启的最大失败次数
         */
        private int maxFailuresPerInterval = 3;
    }

    /**
     * 窗口配置
     */
    @Data
    public static class WindowConfig {
        /**
         * 默认窗口大小（毫秒）
         */
        private long size = 300000; // 5分钟
        
        /**
         * 默认滑动间隔（毫秒）
         */
        private long slide = 60000; // 1分钟
        
        /**
         * 默认会话超时时间（毫秒）
         */
        private long sessionTimeout = 30000; // 30秒
        
        /**
         * 窗口提前触发时间（毫秒）
         */
        private long earlyFiringDelay = 1000;
        
        /**
         * 窗口延迟触发时间（毫秒）
         */
        private long lateFiringDelay = 0;
        
        /**
         * 是否启用延迟数据处理
         */
        private boolean allowLateness = false;
        
        /**
         * 最大延迟时间（毫秒）
         */
        private long maxLateness = 0;
    }

    /**
     * Kafka配置
     */
    @Data
    public static class KafkaConfig {
        /**
         * Kafka服务器地址
         */
        private String bootstrapServers = "localhost:9092";
        
        /**
         * 消费者组ID
         */
        private String consumerGroupId = "flink-compute-group";
        
        /**
         * 自动偏移量重置（earliest/latest/none）
         */
        private String autoOffsetReset = "latest";
        
        /**
         * 是否启用自动提交
         */
        private boolean enableAutoCommit = false;
        
        /**
         * 请求超时时间（毫秒）
         */
        private int requestTimeoutMs = 30000;
        
        /**
         * 会话超时时间（毫秒）
         */
        private int sessionTimeoutMs = 10000;
        
        /**
         * 心跳间隔时间（毫秒）
         */
        private int heartbeatIntervalMs = 3000;
        
        /**
         * 获取记录超时时间（毫秒）
         */
        private int fetchMaxWaitMs = 500;
        
        /**
         * 最小获取字节数
         */
        private int fetchMinBytes = 1;
        
        /**
         * 最大获取字节数
         */
        private int fetchMaxBytes = 52428800; // 50MB
    }

    /**
     * 数据库配置
     */
    @Data
    public static class DatabaseConfig {
        /**
         * 结果表前缀
         */
        private String resultTablePrefix = "stat_result_";
        
        /**
         * 批量写入大小
         */
        private int batchSize = 1000;
        
        /**
         * 批量写入间隔（毫秒）
         */
        private int batchIntervalMs = 5000;
        
        /**
         * 最大重试次数
         */
        private int maxRetries = 3;
        
        /**
         * 重试间隔（毫秒）
         */
        private int retryIntervalMs = 1000;
        
        /**
         * 连接超时时间（秒）
         */
        private int connectionTimeout = 30;
        
        /**
         * 查询超时时间（秒）
         */
        private int queryTimeout = 60;
        
        /**
         * 数据保留天数
         */
        private int dataRetentionDays = 90;
    }

    /**
     * 性能调优配置
     */
    @Data
    public static class PerformanceConfig {
        /**
         * 网络缓冲区数量
         */
        private int networkBuffers = 2048;
        
        /**
         * 网络缓冲区大小（字节）
         */
        private int networkBufferSize = 32768; // 32KB
        
        /**
         * 任务槽数量
         */
        private int taskSlots = 4;
        
        /**
         * 每个任务槽的CPU核心数
         */
        private double cpusPerTaskSlot = 1.0;
        
        /**
         * 每个任务槽的内存大小（MB）
         */
        private int memoryPerTaskSlot = 1024;
        
        /**
         * 对象重用
         */
        private boolean objectReuse = true;
        
        /**
         * 自动优化
         */
        private boolean autoOptimization = true;
        
        /**
         * 并行度优化
         */
        private boolean parallelismOptimization = true;
        
        /**
         * 内存优化
         */
        private boolean memoryOptimization = true;
        
        /**
         * 网络优化
         */
        private boolean networkOptimization = true;
    }

    /**
     * 初始化配置
     */
    @PostConstruct
    public void init() {
        log.info("Flink计算配置初始化完成");
        log.info("作业配置: parallelism={}, maxParallelism={}, timeoutMinutes={}", 
                job.getParallelism(), job.getMaxParallelism(), job.getTimeoutMinutes());
        log.info("检查点配置: enabled={}, interval={}, timeout={}", 
                checkpoint.isEnabled(), checkpoint.getInterval(), checkpoint.getTimeout());
        log.info("状态后端配置: type={}, checkpointDir={}", 
                stateBackend.getType(), stateBackend.getCheckpointDir());
        log.info("重启策略配置: strategy={}, maxRestarts={}, delayBetweenRestarts={}", 
                restart.getStrategy(), restart.getMaxRestarts(), restart.getDelayBetweenRestarts());
        log.info("窗口配置: size={}, slide={}", 
                window.getSize(), window.getSlide());
        log.info("Kafka配置: bootstrapServers={}, consumerGroupId={}", 
                kafka.getBootstrapServers(), kafka.getConsumerGroupId());
        log.info("数据库配置: batchSize={}, dataRetentionDays={}", 
                database.getBatchSize(), database.getDataRetentionDays());
        log.info("性能配置: taskSlots={}, cpusPerTaskSlot={}, memoryPerTaskSlot={}", 
                performance.getTaskSlots(), performance.getCpusPerTaskSlot(), performance.getMemoryPerTaskSlot());
    }
}